package com.jinghang.logProject.App_yarn

import com.jinghang.logProject.utils.DateUtils
import org.apache.spark.sql.SparkSession

/**
  * 第一步清洗数据：抽取出我们所需要的某列的数据（日期 ，url,流量，ip）
  */
object _010_SparkStatFormatApp {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession
      .builder()
      .appName("SparkStatFormatJob1")
      .getOrCreate()

    //根据文件路径获取资源
    val path = args(0)
    var sourceRDD = spark.sparkContext.textFile(path)
    //打印输出前十条数据
    //sourceRDD.take(10).foreach(println)


    sourceRDD.map(
      //对每一行数据进行处理
      line => {
        val splits = line.split(" ")
        println(splits)
        val ip = splits(0)
        /*
          * 原始日志的第三个和第四个字段拼接起来就是完整的访问时间：
          * [10/Nov/2016:00:01:02 +0800] ==> yyyy-MM-dd HH:mm:ss
          */
        val time = splits(3) + " " + splits(4)
        val time_format = DateUtils.parse(time)
        //url去掉 " 单引号
        val url = splits(11).replace("\"", "")
        //通信量 流量
        val traffic = splits(9)
        //把数据打印出来
        // (ip, DateUtils.parse(time),url,trafic)

        val log_format = time_format + "\t" + url + "\t" + traffic + "\t" + ip
        log_format
      }
    )
      .saveAsTextFile(args(1))

    spark.stop()

  }

}
